package com.sh.data.engine.domain.integration.documentCapture.service.impl;

import cn.hutool.extra.spring.SpringUtil;
import com.google.common.collect.Lists;
import com.sh.data.engine.common.constants.Constants;
import com.sh.data.engine.domain.base.service.LogService;
import com.sh.data.engine.domain.integration.documentCapture.service.DocumentCaptureService;
import com.sh.data.engine.infrastructure.config.DataXConfig;
import com.sh.data.engine.repository.dao.integration.documentCapture.entity.DocumentCaptureEntity;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.io.FileUtils;

import java.io.*;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * @author: mengzheng.mhc
 * @date: 2024/6/19 10:01
 */
@Data
@Slf4j
public class DocumentCaptureRunnable implements Runnable {

    private static final String SKIP = "in ch.qos.logback";

    private String jsonScript;

    private String dataxJsonPath;

    private DocumentCaptureEntity documentCaptureEntity;

    private String logDir;

    private LogService logService = SpringUtil.getBean(LogService.class);

    private DocumentCaptureService documentCaptureService =
        SpringUtil.getBean(DocumentCaptureService.class);

    private DataXConfig dataXConfig = SpringUtil.getBean(DataXConfig.class);

    public DocumentCaptureRunnable(
        String jsonScript,
        String dataxJsonPath,
        DocumentCaptureEntity documentCaptureEntity,
        String logDir) {
        this.jsonScript = jsonScript;
        this.dataxJsonPath = dataxJsonPath;
        this.documentCaptureEntity = documentCaptureEntity;
        this.logDir = logDir;
    }

    @Override
    public void run() {
        try {
            File dataxFile = new File(dataxJsonPath);
            FileUtils.writeStringToFile(dataxFile, jsonScript, "UTF-8");

            String cmd = dataXConfig.getCmd(dataxJsonPath);

            log.info("cmd:{}", cmd);

            Process process = Runtime.getRuntime().exec(cmd);

            CountDownLatch latch = new CountDownLatch(2);
            read(process.getInputStream(), documentCaptureEntity.getId(), latch);

            read(process.getErrorStream(), documentCaptureEntity.getId(), latch);

            int code = process.waitFor();
            log.info("exitCode:{}", code);

            String finishedLog = String.format("文本采集:[%s]结束执行", documentCaptureEntity.getTargetTblName());

            String subTaskJobLogPath =
                String.format(
                    "%s/%s/log/%s.log", logDir, Constants.TEXT_EXPORT_LOG, documentCaptureEntity.getId());

            if (code == 0) {
                latch.await();
                this.writeLog(Lists.newArrayList(finishedLog), documentCaptureEntity.getId());
                // 任务执行完成后设置任务为已完成
                documentCaptureEntity.setState(0);
                // 设置上传成功的条数 这个条数从log中取的。
                int successNum = 0;

                int recordTotal = 0;
                int failNum = 0;
                List<String> read = logService.readNew(subTaskJobLogPath, 0);
                if (read != null) {
                    for (String readStr : read) {
                        if (readStr.startsWith("读出记录总数")) {
                            recordTotal = getNumInStr(readStr);
                        }
                        if (readStr.startsWith("读写失败总数")) {
                            failNum = getNumInStr(readStr);
                        }
                    }
                    // 写入成功上传条数
                    documentCaptureEntity.setUploadNum(recordTotal - failNum);
                }
            } else {
                // 任务执行返回其他错误码直接设置失败
                documentCaptureEntity.setState(1);
            }
            documentCaptureService.updateDocCap(documentCaptureEntity);

        } catch (Exception e) {
            log.error("文件采集同步任务失败" + e);
        }
    }

    private void read(InputStream in, Long docCapId, CountDownLatch latch) {
        new Thread(
            () -> {
                try (BufferedReader reader = new BufferedReader(new InputStreamReader(in, "UTF-8"))) {
                    String line;
                    while ((line = reader.readLine()) != null) {
                        if (line.contains(SKIP)) {
                            continue;
                        }
                        List<String> logs = Lists.newArrayList(line);
                        writeLog(logs, docCapId);
                    }
                } catch (IOException e) {
                    log.error(e.getMessage(), e);
                } finally {
                    latch.countDown();
                }
            })
            .start();
    }

    public void writeLog(List<String> lines, Long docCapId) {
        if (CollectionUtils.isEmpty(lines)) {
            return;
        }

        String subTaskJobLogPath =
            String.format("%s/%s/log/%s.log", logDir, Constants.TEXT_EXPORT_LOG, docCapId);

        logService.write(subTaskJobLogPath, lines);
    }

    /**
     * @description : 提取字符串中的数字
     * @author : tinglan.ys
     * @date : 2022/3/21
     */
    private int getNumInStr(String str) {
        String regEx = "[^0-9]";
        Pattern p = Pattern.compile(regEx);
        Matcher m = p.matcher(str);
        return Integer.parseInt(m.replaceAll("").trim());
    }
}
